package com.shortLinked.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.shortLinked.project.common.constant.RedisKeyConstant;
import com.shortLinked.project.common.constant.ShortLinkConstant;
import com.shortLinked.project.common.convention.exception.ServiceException;
import com.shortLinked.project.dao.entity.*;
import com.shortLinked.project.dao.mapper.*;
import com.shortLinked.project.dto.biz.ShortLinkStatsRecordDTO;
import com.shortLinked.project.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * 短链接监控状态保持消息队列消费者 - 自动拿取进行消费
 */
@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "${rocketmq.producer.topic}",
        consumerGroup = "${rocketmq.consumer.group}"
)
public class ShortLinkStatsSaveConsumer implements RocketMQListener<Map<String, String>> {
    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkGoToMapper shortLinkGotoMapper;
    private final RedissonClient redissonClient;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    // 获取地区位置的请求 key
    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;

    // 接收MQ中的消息
    @Transactional(rollbackFor = Throwable.class)
    public void onMessage(Map<String, String> producerMap) {
        String keys = producerMap.get("keys");
        if (messageQueueIdempotentHandler.isMessageBeingConsumed(keys)) {
            if (messageQueueIdempotentHandler.isAccomplish(keys)) {
                return;
            }
            throw new ServiceException("消息未完成消费流程，需要消息队列重新消费");
        }
        try {
            // 如果消费失败了，消息 ID 已经插进入了  - 解决方式
            ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
            actualSaveShortLinkStats(statsRecord);
            // 如果消费没有消费完，就宕机了，解决方案 - 通过增加标志的方式，再去拉取消息进行消费
        } catch (Throwable ex) {
            messageQueueIdempotentHandler.delMessageProcessed(keys);
            log.error("短链接监控统计消息消费异常：" + ex);
            // 出现异常抛出，避免设置该消息已完成状态
            throw ex;
        }
        // 完成消息消费
        messageQueueIdempotentHandler.setAccomplish(keys);
    }


    private void actualSaveShortLinkStats(ShortLinkStatsRecordDTO statsRecord) {
        String fullShortUrl = statsRecord.getFullShortUrl();
        // 引入读写锁，拿到读锁，其他读锁也可以获取，但是写锁无法获取 - 这里加读锁
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(RedisKeyConstant.LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();
        // 加入了 mq 之后，就不再需要再放入延迟队列了，mq 获取监控统计消息是每个一秒获取一条消息进行消费，远远大于你修改短链接所需要的时间
        rLock.lock();
        try {
            // 此时去数据库里面实现的拿取 fullShortUrl 的 gid
            LambdaQueryWrapper<ShortLinkGoToDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGoToDO.class)
                    .eq(ShortLinkGoToDO::getFullShortUrl, statsRecord.getFullShortUrl());
            ShortLinkGoToDO shortLinkGoToDO = shortLinkGotoMapper.selectOne(queryWrapper);
            String gid = shortLinkGoToDO.getGid();
            // 获取当前小时时间以及星期时间
            Date currentDate = statsRecord.getCurrentDate();
            int hour = DateUtil.hour(currentDate, true);
            Week week = DateUtil.dayOfWeekEnum(currentDate);
            int weekValue = week.getIso8601Value();
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .hour(hour)
                    .weekday(weekValue)
                    .date(currentDate)
                    .build();
            linkAccessStatsMapper.linkAccessStats(linkAccessStatsDO);

            // 对地域访问进行一个统计
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", statsLocaleAmapKey);
            localeParamMap.put("ip", statsRecord.getRemoteAddr());
            String localeResultStr = HttpUtil.get(ShortLinkConstant.AMAP_LOCALE_URL, localeParamMap);
            // 对结果进行序列化
            JSONObject localeResultObj = JSON.parseObject(localeResultStr);
            String infocode = localeResultObj.getString("infocode");
            LinkLocaleStatsDO localeStatsDO;
            String actualProvince = "未知";   // 真实省份
            String actualCity = "未知";       // 真实城市
            if (StrUtil.isNotBlank(infocode) && StrUtil.equals(infocode, "10000")) {
                String province = localeResultObj.getString("province"); // 省份
                Boolean unKownFlag = StrUtil.equals("[]", province);
                localeStatsDO = LinkLocaleStatsDO.builder()
                        .province(actualProvince = unKownFlag ? actualProvince : province)
                        .city(actualCity = unKownFlag ? actualCity : localeResultObj.getString("city"))   // 城市名称
                        .adcode(unKownFlag ? "未知" : localeResultObj.getString("adcode"))  // 城市 adcode 编码
                        .fullShortUrl(fullShortUrl)
                        .cnt(1)
                        .country("中国")
                        .date(currentDate)
                        .build();
                linkLocaleStatsMapper.linkLocaleStats(localeStatsDO);
            }

            // 短链接操作系统访问监控
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecord.getOs())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);

            // 浏览器访问统计
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);

            // 设备访问统计
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .cnt(1)
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);

            // 网络访问统计
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .cnt(1)
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);

            // 访问日志统计
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .user(statsRecord.getUv())
                    .ip(statsRecord.getRemoteAddr())
                    .browser(statsRecord.getBrowser())
                    .os(statsRecord.getOs())
                    .network(statsRecord.getNetwork())
                    .device(statsRecord.getDevice())
                    .locale(StrUtil.join("-", "中国", actualProvince, actualCity))
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } catch (Throwable ex) {
            log.error("短链接访问量统计异常 - " + ex);
        } finally {
            rLock.unlock();
        }
    }
}
